use async_trait::async_trait;
use fluviox_shared::{Result, FluvioXError, models::DataRecord};
use crate::traits::{SmartModule, Transform, SmartModuleType, ProcessResult};
use serde_json::Value;
use std::collections::HashMap;
use tracing::{info, warn};

/// 字段映射转换器
/// 用于重命名、添加、删除字段
pub struct FieldMapper {
    name: String,
    config: Option<FieldMapperConfig>,
}

/// 字段映射配置
#[derive(Debug, Clone)]
pub struct FieldMapperConfig {
    /// 字段重命名映射 (old_name -> new_name)
    pub field_mappings: HashMap<String, String>,
    /// 要删除的字段
    pub fields_to_remove: Vec<String>,
    /// 要添加的字段 (field_name -> value)
    pub fields_to_add: HashMap<String, Value>,
    /// 默认值映射 (field_name -> default_value)
    pub default_values: HashMap<String, Value>,
    /// 是否保留未映射的字段
    pub keep_unmapped: bool,
}

impl FieldMapper {
    pub fn new() -> Self {
        Self {
            name: "field-mapper".to_string(),
            config: None,
        }
    }
    
    /// 应用字段映射
    fn apply_mapping(&self, mut data: Value, config: &FieldMapperConfig) -> Result<Value> {
        if let Value::Object(ref mut obj) = data {
            let mut new_obj = serde_json::Map::new();
            
            // 处理现有字段
            for (key, value) in obj.iter() {
                // 检查是否要删除
                if config.fields_to_remove.contains(key) {
                    continue;
                }
                
                // 检查是否要重命名
                let new_key = config.field_mappings.get(key).unwrap_or(key);
                
                // 如果不保留未映射字段，只保留映射的字段
                if !config.keep_unmapped && !config.field_mappings.contains_key(key) {
                    continue;
                }
                
                new_obj.insert(new_key.clone(), value.clone());
            }
            
            // 添加新字段
            for (key, value) in &config.fields_to_add {
                new_obj.insert(key.clone(), value.clone());
            }
            
            // 添加默认值（如果字段不存在）
            for (key, default_value) in &config.default_values {
                if !new_obj.contains_key(key) {
                    new_obj.insert(key.clone(), default_value.clone());
                }
            }
            
            Ok(Value::Object(new_obj))
        } else {
            // 非对象类型，直接返回
            Ok(data)
        }
    }
}

#[async_trait]
impl SmartModule for FieldMapper {
    fn name(&self) -> &str {
        &self.name
    }
    
    fn version(&self) -> &str {
        "1.0.0"
    }
    
    fn description(&self) -> &str {
        "字段映射转换器，支持字段重命名、添加、删除和默认值设置"
    }
    
    fn module_type(&self) -> SmartModuleType {
        SmartModuleType::Transform
    }
    
    async fn initialize(&mut self, config: Value) -> Result<()> {
        info!("初始化字段映射转换器");
        
        let field_mappings = config.get("field_mappings")
            .and_then(|v| v.as_object())
            .map(|obj| {
                obj.iter()
                    .filter_map(|(k, v)| v.as_str().map(|s| (k.clone(), s.to_string())))
                    .collect()
            })
            .unwrap_or_default();
        
        let fields_to_remove = config.get("fields_to_remove")
            .and_then(|v| v.as_array())
            .map(|arr| {
                arr.iter()
                    .filter_map(|v| v.as_str().map(|s| s.to_string()))
                    .collect()
            })
            .unwrap_or_default();
        
        let fields_to_add = config.get("fields_to_add")
            .and_then(|v| v.as_object())
            .map(|obj| {
                obj.iter()
                    .map(|(k, v)| (k.clone(), v.clone()))
                    .collect()
            })
            .unwrap_or_default();
        
        let default_values = config.get("default_values")
            .and_then(|v| v.as_object())
            .map(|obj| {
                obj.iter()
                    .map(|(k, v)| (k.clone(), v.clone()))
                    .collect()
            })
            .unwrap_or_default();
        
        let keep_unmapped = config.get("keep_unmapped")
            .and_then(|v| v.as_bool())
            .unwrap_or(true);
        
        self.config = Some(FieldMapperConfig {
            field_mappings,
            fields_to_remove,
            fields_to_add,
            default_values,
            keep_unmapped,
        });
        
        info!("字段映射转换器初始化完成");
        Ok(())
    }
    
    async fn process(&mut self, mut record: DataRecord) -> Result<Vec<ProcessResult>> {
        let config = self.config.as_ref()
            .ok_or_else(|| FluvioXError::Internal("转换器未初始化".to_string()))?;
        
        // 应用字段映射
        record.data = self.apply_mapping(record.data, config)?;
        
        Ok(vec![ProcessResult::Transform(record)])
    }
}

#[async_trait]
impl Transform for FieldMapper {
    async fn transform(&self, mut record: DataRecord) -> Result<DataRecord> {
        let config = self.config.as_ref()
            .ok_or_else(|| FluvioXError::Internal("转换器未初始化".to_string()))?;
        
        record.data = self.apply_mapping(record.data, config)?;
        Ok(record)
    }
}

impl Default for FieldMapper {
    fn default() -> Self {
        Self::new()
    }
}
